package com.pisolution.scheduler.runtime.master.processor;

import com.pisolution.scheduler.core.utils.JsonUtils;
import com.pisolution.scheduler.datasource.instance.ExecutionState;
import com.pisolution.scheduler.remote.command.Command;
import com.pisolution.scheduler.remote.command.CommandType;
import com.pisolution.scheduler.remote.processor.NettyRequestProcessor;
import com.pisolution.scheduler.remote.utils.Host;
import com.pisolution.scheduler.runtime.SchedulerContext;
import com.pisolution.scheduler.runtime.master.cache.TaskInstanceCache;
import com.pisolution.scheduler.runtime.master.cache.TaskInstanceCacheManager;
import com.pisolution.scheduler.runtime.worker.processor.TaskAckCommand;

import com.google.common.base.Preconditions;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 处理工作节点反馈给管理节点的任务确认请求 请求处理逻辑详见具体的处理方法描述.
 *
 * @see #process(Channel, Command)
 * @author sora
 * @version 0.1.0
 */
public class TaskAckProcessor implements NettyRequestProcessor {
    private static final Logger LOGGER = LoggerFactory.getLogger(TaskAckProcessor.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于更新任务实例状态. */
    private final TaskStateUpdater updater;

    /** 用于缓存任务实例对象. */
    private final TaskInstanceCache cache;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    public TaskAckProcessor() {
        this.updater = SchedulerContext.get(TaskStateUpdater.class);
        this.cache = SchedulerContext.get(TaskInstanceCacheManager.class);
    }

    // -------------------------------------------------------------------------------------------------
    //  请求处理
    // -------------------------------------------------------------------------------------------------

    @Override
    public void process(Channel channel, Command command) {
        // 1. 校验接收到的指令类型是否与待处理的指令类型一致.
        Preconditions.checkArgument(
            CommandType.TASK_ACK_REQUEST == command.getType(),
            "Received unrecognized command with type %s",
            command.getType()
        );

        // 2. 反序列化指令内容.
        final TaskAckCommand cmd = JsonUtils.deserialize(command.getBody(), TaskAckCommand.class);

        LOGGER.info("Received the interaction command {}.", cmd);

        // 3. 从指令中解析出任务实例并缓存 构建响应体并追加到.
        this.cache.cache(cmd);
        this.updater.addEvent(new TaskStateEvent(
            cmd.getId(),
            Host.of(channel).getAddress(),
            channel,
            cmd.getExecutePath(),
            cmd.getLogPath(),
            ExecutionState.of(cmd.getStatus()),
            cmd.getStartTime()
        ));
    }
}
